/*
 * Copyright 2017 StreamSets Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.streamsets.pipeline.stage.processor.kudulookup;

import com.streamsets.pipeline.api.Config;
import com.streamsets.pipeline.api.StageException;
import com.streamsets.pipeline.api.StageUpgrader;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.stage.common.MissingValuesBehavior;
import com.streamsets.pipeline.stage.destination.kudu.KuduConfigBean;

import java.util.ArrayList;
import java.util.List;

public class KuduProcessorUpgrader implements StageUpgrader {

  @Override
  public List<Config> upgrade(
      String library, String stageName, String stageInstance, int fromVersion, int toVersion, List<Config> configs
  ) throws StageException {
    switch (fromVersion) {
      case 1:
        upgradeV1ToV2(configs);
        if (toVersion == 2) {
          break;
        }
        // fall through
      case 2:
        upgradeV2ToV3(configs);
        if (toVersion == 3) {
          break;
        }
        // fall through
      case 3:
        upgradeV3ToV4(configs);
        break;
      default:
        throw new IllegalStateException(Utils.format("Unexpected fromVersion {}", fromVersion));
    }
    return configs;
  }

  private static void upgradeV1ToV2(List<Config> configs) {
    configs.add(new Config("conf.missingLookupBehavior", MissingValuesBehavior.SEND_TO_ERROR));
  }

  private static void upgradeV2ToV3(List<Config> configs) {
    configs.add(new Config("conf.adminOperationTimeout", 30000));
    configs.add(new Config("conf.numWorkers", 0)); // use default
  }

  private static void upgradeV3ToV4(List<Config> configs) {
    List<Config> configsToRemove = new ArrayList<>();
    List<Config> configsToAdd = new ArrayList<>();
    configs.forEach(config -> {
      if ((KuduLookupConfig.CONF_PREFIX + "adminOperationTimeout").equals(config.getName())) {
        configsToAdd.add(new Config(KuduLookupConfig.CONNECTION_PREFIX + "adminOperationTimeout" , config.getValue()));
        configsToRemove.add(config);
      }
      if ((KuduLookupConfig.CONF_PREFIX + "kuduMaster").equals(config.getName())) {
        configsToAdd.add(new Config(KuduLookupConfig.CONNECTION_PREFIX +  "kuduMaster" , config.getValue()));
        configsToRemove.add(config);
      }
      if ((KuduLookupConfig.CONF_PREFIX + "numWorkers").equals(config.getName())) {
        configsToAdd.add(new Config(KuduLookupConfig.CONNECTION_PREFIX + "numWorkers" , config.getValue()));
        configsToRemove.add(config);
      }
      if ((KuduLookupConfig.CONF_PREFIX + "operationTimeout").equals(config.getName())) {
        configsToAdd.add(new Config(KuduLookupConfig.CONNECTION_PREFIX + "operationTimeout" , config.getValue()));
        configsToRemove.add(config);
      }
    });
    configs.removeAll(configsToRemove);
    configs.addAll(configsToAdd);
  }
}
